package com.junjie.blibli.service.webSocket;

import com.alibaba.fastjson.JSONObject;
import com.junjie.blibli.Constant.RocketMqConstant;
import com.junjie.blibli.Entity.Danmu;
import com.junjie.blibli.service.DanmuService;
import com.junjie.blibli.util.RocketMQUtil;
import com.junjie.blibli.util.TokenUtil;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * websocket服务类
 * 只要加了@ServerEndpoint()注解，就代表是服务类了
 * 加上路径就有点类似controller，前端就知道上哪找webSocket
 * 这里加上{token}效果类似于@PathVariable，因为我们需要获取userId，就要拦截Token
 */
@Component
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {
    //获取跟这个类相关的logger，日志
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    //用户ID，通过拦截token获取，在进行webSocket连接的时候
    private Long userId;

    /**
     * 当前长链接人数，直接用static全局变量
     * AtomicInteger是java提供的一个原子性操作的一个类，来保证线程安全
     * 原子性操作用于高并发操作，因为高并发的时候线程不一定是安全的
     */
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    /**
     * ConcurrentHashMap也是保证线程安全的类
     * 作用是保存每个客户端对应的webSocketService
     * 当每个客户端进来后，高效获取相应客户端的webSocket服务
     * 这里用public修饰，因为其他地方要用
     *
     * 那为什么每个客户端都需要单独的webSocketService?
     * 你未必每个用户进来都是用一个webSocket啊，怎么可能
     * 而且因为Spring默认是单例模式，我们后续需要使用多例
     */
    public static final ConcurrentHashMap<String,WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    /**
     * 服务端和客户端进行通信的会话
     * 这个会话主要是用来进行长链接通信的
     * 当我们有一个客户端进来了，链接成功了
     * 我们就会保存一个跟这个客户端关联的session
     * 这个session在下次客户端对我们进行信息传输的时候，我们就可以通过session与客户端进行通信
     * 比如说服务端想推送消息给客户端，这时候就可以把session拿过来
     * 流程：首先通过客户端的唯一标识从WEBSOCKET_MAP取到webSocketService，就可以取到Session
     * 然后就可以通信了，具体通信就是调用相关的方法
     */
    private Session session;

    //即WEBSOCKET_MAP中的key
    private String sessionId;

    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }


    /**
     * 因为webSocket是多例，不能直接用@Autowire注入，这个注入是单例的，只会注入第一个webSocket
     * 静态变量就可以全局共享了，然后我们获取上下文对象后就可以自定义获取bean了
     */
    private static ApplicationContext APPLICATION_CONTEXT;

    //这个静态方法是提供给启动类那里用的，直接给静态变量赋值
    public static void setApplicationContext(ApplicationContext applicationContext){
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }

    /**
     * 建立连接方法
     * @param session 这个session是服务端与客户端自动建立起来的
     * @OnOpen 注解是默认建立起websocket要调用的方法
     * @PathParam 这个注解类似于@PathVariable
     */
    @OnOpen
    public void openConnection(Session session, @PathParam("token") String token){
        //因为游客也是可以看视频和弹幕的，所以要加一个try-catch，不需要处理异常
        try{
            //破解token获取userId
            this.userId = TokenUtil.verifyToken(token);
        }catch (Exception e){}
        //通过全局上下文对象，获得对应的bean
        RedisTemplate<String,String> redisTemplate = (RedisTemplate<String,String>) WebSocketService
                .APPLICATION_CONTEXT.getBean("RedisTemplate");
        //给本地sessionId和session赋值
        this.sessionId = session.getId();
        this.session = session;
        //判断一下WEBSOCKET_MAP之前有没有存储过sessionId，有就先去掉，再添加新的sessionId
        if(WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId,this);
        }else{
            //没有的话直接添加
            WEBSOCKET_MAP.put(sessionId,this);
            //在线人数+1，因为是客户端第一次连接服务端
            //这里getAndIncrement的意思是先获取到值再+1
            ONLINE_COUNT.getAndIncrement();
        }
        //get方法就是返回原始的Int类型
        logger.info("用户连接成功："+sessionId+"，当前在线人数为："+ONLINE_COUNT.get());
        //通知前端或者客户端连接成功了，让他们干其他操作
        try {
            this.sendMessage("0");
        } catch (Exception e){
            logger.error("连接异常");
        }
    }

    /**
     * 关闭连接方法
     * 关闭通常是什么情况呢
     * 一、服务端直接断了，这时候就刷新不出来
     * 二、浏览器里面关闭了当前页面
     * @OnOpen 注解是默认建立起websocket要调用的方法
     */
    @OnClose
    public void closeConnection(){
        if(WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        logger.info("用户退出:"+sessionId+",当前在线人数为："+ONLINE_COUNT.get());
    }

    /**
     * 比如说前端把消息发给服务端就会调用这个方法
     * @param message 服务端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message){
        logger.info("用户信息" + sessionId +"，报文：" + message);
        if(!StringUtil.isNullOrEmpty(message)){
            try{
                //当某个客户端发消息给服务端后，服务端需要群发消息，直接遍历整个Map就行了
                for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()) {
                    //拿出每个人的webSocketService
                    WebSocketService webSocketService = entry.getValue();
                    //引入MQ生产者，接下来的事情交给MQ去做
                    DefaultMQProducer danmusProducer = (DefaultMQProducer)
                            APPLICATION_CONTEXT.getBean("damusProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message",message);
                    jsonObject.put("sessionId",webSocketService.getSessionId());
                    //创建消息
                    Message msg = new Message(RocketMqConstant.TOPIC_DANMUS
                            ,jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    //异步发送消息
                    RocketMQUtil.asyncSendMsg(danmusProducer,msg);
                }
                if(this.userId != null){
                    //将该信息转换成实体类，异步存到数据库
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    //设置一些属性
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService) APPLICATION_CONTEXT.getBean("danmuService");
                    danmuService.asyncAddDanmu(danmu);
                    //弹幕存到redis里面，这里可以不用异步，因为redis非常快
                    //但是如果redis是分布式，就要考虑用分布式锁
                    danmuService.addDanmusToRedis(danmu);
                }

            }catch (Exception e){
                logger.error("弹幕接收出现问题");
                e.printStackTrace();
            }
        }
    }

    /**
     * 发生错误之后进行处理
     * @param error 出现的异常
     */
    @OnError
    public void OnError(Throwable error){

    }

    /**
     * 服务端给客户端发送消息
     * @param message 具体消息
     */
    public void sendMessage(String message) throws Exception {
        //直接用session发送消息
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 统计在线人数
     * 使用定时任务，每五秒更新一次
     */
    @Scheduled(fixedRate = 5000)
    private void noticeOnlineCount() throws Exception{
        //遍历每个用户，挨个推送信息
        for (Map.Entry<String, WebSocketService> entry : WebSocketService.WEBSOCKET_MAP.entrySet()) {
            WebSocketService webSocketService = entry.getValue();
            if(webSocketService.session.isOpen()){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount",ONLINE_COUNT.get());
                jsonObject.put("msg","当前在线人数为："+ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }
}
